package com.btcode.netty.server.handler;

import com.btcode.common.MyLog;
import com.btcode.log.ILog;
import com.btcode.netty.server.session.IoSession;
import com.btcode.netty.server.session.SessionManager;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.internal.OutOfDirectMemoryError;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 1.消息日志（同时支持字符串协议与字节码协议）
 * 2.消息拦截（可选，需覆盖实现shouldIntercept，可以根据消息内容或者会话属性进行业务拦截）
 * 3.基础的消息频率控制（可选，需覆盖实现getRtRate，返回数组第一个值大于0生效
 * 4.业务消息处理方法（必须实现）
 */
public abstract class AbstractJobIntercepter extends ChannelInboundHandlerAdapter {
    protected final ILog log = MyLog.getInstance().getLogger(AbstractJobIntercepter.class);
    private Map<String, AtomicInteger> rtCountMap = new HashMap<>();
    private volatile long rtStartSeconds = System.currentTimeMillis() / 1000;

    @Override
    public final void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            IoSession session = SessionManager.INSTANCE.getSessionByChannel(ctx.channel());
            logMsg((ByteBuf) msg, session);
            if (!shouldIntercept((ByteBuf) msg, session)) {
                if (getRtRate()[0] <= 0 || isDataRateNormal(session, (ByteBuf) msg)) {
                    doBizWithMsg((ByteBuf) msg, session);
                }
            }
        }
        catch (Exception e) {
            log.error(SessionManager.INSTANCE.getSessionByChannel(ctx.channel()), e);
        }
    }

    @Override
    public final void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error(SessionManager.INSTANCE.getSessionByChannel(ctx.channel()), cause);
        if (cause instanceof OutOfDirectMemoryError) {
            SessionManager.INSTANCE.unRegisterUserContext(ctx.channel());
        }
    }

    /**
     * 用于实际业务处理，耗时操作建议使用mq，并按业务数据类型对消息分类分队列处理
     * MqKit.getInstance().productMsg("MQ_SEND", msg, false);
     *
     * @param msg 解码后的消息体，简易字符串处理((ByteBuf) msgDecodeOut).toString(Charset.defaultCharset())
     */
    public abstract void doBizWithMsg(ByteBuf msg, IoSession session);

    /**
     * 日志记录，默认按字符串协议打印，可以覆盖自定义如字节协议快速解码方法MsgDecoderWithSessionCtrl.getByteBufHexStr
     *
     * @param msg 解码后的消息体，简易字符串处理((ByteBuf) msgDecodeOut).toString(Charset.defaultCharset())
     */
    public void logMsg(ByteBuf msg, IoSession session) {
        log.info(AbstractHrtDecoder.getByteBufString(msg));
    }

    /**
     * 根据消息内容或者会话属性进行业务拦截，比如实时频率过快，默认不拦截，开启需要覆盖实现getRtRate；
     * 或者是黑名单白名单处理 & 状态设置 com.btcode.netty.server.session.util.StationService
     *
     * @param msg     解码后的消息体，简易字符串处理((ByteBuf) msgDecodeOut).toString(Charset.defaultCharset())
     * @param session 连接会话，可用于快速MN、IP、活跃时间等获取，执行具体的消息拦截和业务处理，如果没有按要求实现解码器getMnFromMsg方法，session可能为null
     * @return
     */
    public boolean shouldIntercept(ByteBuf msg, IoSession session) {
        return false;
    }

    private boolean isDataRateNormal(IoSession session, ByteBuf msgDecodeOut) {
        try {
            if (!shouldFilterRate(msgDecodeOut)) {
                return true;
            }
            if (System.currentTimeMillis() / 1000 - rtStartSeconds > 86400) {
                rtStartSeconds = System.currentTimeMillis() / 1000;
                rtCountMap = new HashMap<>();
            }
            String mn = session.getUserId();
            if (!rtCountMap.containsKey(mn)) {
                rtCountMap.put(mn, new AtomicInteger(0));
            }
            int nowCount = rtCountMap.get(mn).incrementAndGet();
            if (nowCount > 1) {//重置时第一条不计算
                double rate = nowCount * 60 / (System.currentTimeMillis() / 1000.0 - rtStartSeconds);
                if (nowCount > getRtRate()[0] || rate > getRtRate()[1]) {
                    log.info(String.format("%s数据超标！每分钟条数：%s", mn, rate));
                    return false;
                }
            }
        }
        catch (Exception e) {
            log.error(null, e);
        }
        return true;
    }

    /**
     * 数据频率处理的时候，过滤数据类型，默认全部，可覆盖
     */
    public boolean shouldFilterRate(ByteBuf msgDecodeOut) {
        return true;
    }

    // 2880,2 按半分钟一条实时，每天正常2880条，正常情况实时数据一分钟最多2条
    private static final double[] RR = new double[]{0, 1};

    /**
     * 数据频率控制，可扩展到redis动态控制每个ID连接：每日最大数量 & 每分钟允许最大数量
     * 默认不做控制
     *
     * @return
     */
    public double[] getRtRate() {
        return RR;
    }
}